package kafka.loganalysis;

import kafka.admin.AdminUtils;
import kafka.utils.ZKStringSerializer$;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import scala.Tuple2;
import scala.collection.Iterator;
import scala.collection.Map;

import java.util.Properties;

/**
 * @author: pizhihui
 * @datae: 2017-11-08
 */
public class KafkaTopicUtils {

    private static final String DEFAULT_ZK_CONNECT = "hfcdh-02:2181,hfcdh-03:2181,hfcdh-04:2181/kafka";
    private static final int DEFAULT_ZK_SESSION_TIMEOUT_MS = 10 * 1000;
    private static final int DEFAULT_ZK_CONNECTION_TIMEOUT_MS = 8 * 1000;



    public static void createTopic(String topic) {
        createTopic(topic, 3, 3, new Properties());
    }

    public static void createTopic(String topic, int partitions, int replication) {
        createTopic(topic, partitions, replication, new Properties());
    }

    /**
     * 创建topic
     * @param topic
     * @param partitions
     * @param replication
     * @param topicConfig
     * @return
     */
    public static void createTopic(String topic, int partitions, int replication, Properties topicConfig) {

        ZkClient zkClient = new ZkClient(DEFAULT_ZK_CONNECT, DEFAULT_ZK_SESSION_TIMEOUT_MS, DEFAULT_ZK_CONNECTION_TIMEOUT_MS, ZKStringSerializer$.MODULE$);
        AdminUtils.createTopic(zkClient, topic, partitions, replication, topicConfig);
        // 0.10 用如下代码
        //AdminUtils.createTopic(zkUtils, topic, partitions, replication, topicConfig, RackAwareMode.Enforced$.MODULE$);
    }

    /**
     * 判断topic是否存在
     * @param topic
     * @return false不存在, true存在
     */
    public static boolean existsTpic(String topic) {
        ZkClient zkClient = new ZkClient(DEFAULT_ZK_CONNECT, DEFAULT_ZK_SESSION_TIMEOUT_MS, DEFAULT_ZK_CONNECTION_TIMEOUT_MS, ZKStringSerializer$.MODULE$);
        return AdminUtils.topicExists(zkClient, topic);

    }


    /**
     * 列举出所有的topic和配置
     */
    public static void listTopics() {
        ZkClient zkClient = new ZkClient(DEFAULT_ZK_CONNECT, DEFAULT_ZK_SESSION_TIMEOUT_MS, DEFAULT_ZK_CONNECTION_TIMEOUT_MS, ZKStringSerializer$.MODULE$);
        Map<String, Properties> configs = AdminUtils.fetchAllTopicConfigs(zkClient);
        final Iterator it = configs.iterator();
        while (it.hasNext()) {
            final Tuple2<String, Properties> topicConfig = (Tuple2<String, Properties>) it.next();
            final String topic = topicConfig._1;
            final Properties prop = topicConfig._2;
            System.out.println(topic + " :  " + ToStringBuilder.reflectionToString(prop, ToStringStyle.SHORT_PREFIX_STYLE));
        }
        zkClient.close();
    }

    /**
     * 返回某个topic的属性值
     * @param topic
     * @return
     */
    public static Properties getTopicProps(String topic) {
        ZkClient zkClient = new ZkClient(DEFAULT_ZK_CONNECT, DEFAULT_ZK_SESSION_TIMEOUT_MS, DEFAULT_ZK_CONNECTION_TIMEOUT_MS, ZKStringSerializer$.MODULE$);
        try {
            Map<String, Properties> configs = AdminUtils.fetchAllTopicConfigs(zkClient);
            final Iterator it = configs.iterator();
            while (it.hasNext()) {
                final Tuple2<String, Properties> topicConfig = (Tuple2<String, Properties>) it.next();
                final String topicTmp = topicConfig._1;
                final Properties prop = topicConfig._2;
                if (topic.equals(topicTmp)) {
                    return prop;
                }
            }
            return new Properties();
        } finally {
            zkClient.close();
        }
    }

    /**
     * 修改某个topic的属性值
     * @param topic 修改的topic
     * @param props 设定的Properties属性值
     */
    public static void changeTopicConfig(String topic, Properties props) {
        ZkClient zkClient = new ZkClient(DEFAULT_ZK_CONNECT, DEFAULT_ZK_SESSION_TIMEOUT_MS, DEFAULT_ZK_CONNECTION_TIMEOUT_MS, ZKStringSerializer$.MODULE$);
        try {
            Properties topicProps = getTopicProps(topic);
            Properties merged = new Properties();
            merged.putAll(topicProps);
            merged.putAll(props);
            AdminUtils.changeTopicConfig(zkClient, topic, merged);
        } finally {
            zkClient.close();
        }

    }

    public static void main(String[] args) {

        listTopics();

    }

}
